package com.ipinyou.mprofile

import kafka.serializer.StringDecoder
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{ Seconds, StreamingContext }
import org.apache.spark.{ SparkConf, SparkContext }
import java.nio.charset.Charset
import org.apache.spark.rdd.RDD
import java.nio.file.Files
import java.io.File
import org.apache.hadoop.util.Time
import org.apache.spark.streaming.Minutes

/**
 * Created by yuwenbing on 15/12/1.
 */

object CheckpointTest {

  def createContext(checkpointDirectory: String): StreamingContext = {

    // If you do not see this printed, that means the StreamingContext has been loaded
    // from the new checkpoint
    println("-----------------Creating new context")
    val sparkConf = new SparkConf().setAppName("RecoverableNetworkWordCount").setMaster("local[4]")
    // Create the context with a 1 second batch size
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    ssc.checkpoint(checkpointDirectory)

    val brokers = "192.168.145.201:9092," +
      "192.168.145.202:9092," +
      "192.168.145.203:9092," +
      "192.168.145.204:9092," +
      "192.168.145.205:9092," +
      "192.168.145.206:9092," +
      "192.168.145.207:9092," +
      "192.168.145.208:9092," +
      "192.168.145.209:9092," +
      "192.168.145.210:9092";

    println("-----------------" + brokers)

    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
    val topicSet = Set("adv")
    val impDStreamRdd = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicSet)
    val message = impDStreamRdd.map { x =>
      val topic = x._2
      val arr = topic.split("\t", -1)
      (arr(15), arr(6))
    }
    message.print(10)
    ssc
  }

  def main(args: Array[String]): Unit = {
    val checkpointDirectory = "file:///Users/miaoyujia/tmp/checkpoint"
//    val checkpointDirectory = "tmp/checkpoint"
    val ssc = StreamingContext.getOrCreate(checkpointDirectory, () => createContext(checkpointDirectory))
    ssc.start()
    ssc.awaitTermination()
  }

}
